package com.atguigu.flink.chapter08;

import com.atguigu.flink.bean.LoginEvent;
import com.atguigu.flink.util.KKutil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.List;

public class Flink08_Project_Spite_Login {
       public static void main(String[] args) {
               Configuration configuration = new Configuration();
               //web  UI端口
               configuration.setInteger("rest.prot",10000);
               StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
               env.setParallelism(1);

                env.readTextFile("input/LoginLog.csv")
                        .map(line->{
                            String[] datas = line.split(",");
                            return new LoginEvent(
                                    Long.valueOf(datas[0]),
                                    datas[1],
                                    datas[2],
                                    Long.valueOf(datas[3])
                            );
                        })
                        .assignTimestampsAndWatermarks(WatermarkStrategy
                                .<LoginEvent>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner((log,es)->log.getEventTime())
                        )
                        .keyBy(LoginEvent::getUserId)
                        //滑动窗口为2秒，步长为1秒
                       // .window(SlidingEventTimeWindows.of(Time.seconds(2),Time.seconds(1)))
                        .countWindow(2,1)
                        .process(new ProcessWindowFunction<LoginEvent, String, Long, GlobalWindow>() {
                            @Override
                            public void process(Long key,
                                                Context ctx,
                                                Iterable<LoginEvent> elements,
                                                Collector<String> out) throws Exception {
                                List<LoginEvent> list = KKutil.toList(elements);
                                if (list.size()==2) {
                                    LoginEvent e1 = list.get(0);
                                    LoginEvent e2 = list.get(1);

                                    String t1 = e1.getEventType();
                                    String t2 = e2.getEventType();

                                    Long time1 = e1.getEventTime();
                                    Long time2 = e2.getEventTime();

                                    if ("fail".equals(t1) && "fail".equals(t2) && Math.abs(time1-time2)<=2) {

                                        //只能解决基本不乱序问题，乱序过大，还需对时间状态进行分析，后期使用CEP来解决
                                        out.collect("用户"+key + "在恶意登录");
                                    }
                                }
                            }
                        }).print();




               try {
                   env.execute();
               } catch (Exception e) {
                   e.printStackTrace();
               }


           }
}
